package com.bw.gmall.realtime.dws.app;

/**
 * @BelongsProject: gmall_lx
 * @BelongsPackage: com.bw.gmall.realtime.dws.app
 * @Author: cuizexuan
 * @CreateTime: 2025-06-24  14:36
 * @Description: TODO
 * @Version: 1.0
 */
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.bw.realtime_common.base.BaseApp;
import com.bw.realtime_common.bean.TradeSkuOrderBean;
import com.bw.realtime_common.constant.Constant;
import com.bw.realtime_common.function.DorisMapFunction;
import com.bw.realtime_common.util.*;
import com.bw.realtime_common.util.HbaseUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Random;

public class DwsTradeSkuOrderSyncWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderSyncWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.DWS_TRADE_SKU_ORDER_WINDOW,1,10080);
    }
    private boolean isJsonValid(String json) {
        try {
            JSON.parse(json);
            return true;
        } catch (JSONException e) {
            return false;
        }
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1.ETL清洗数据
        SingleOutputStreamOperator<JSONObject> etlStream = dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {


                try {
                    // 先验证是否为JSON格式
                    if (s != null && isJSONValid(s)) {
                        JSONObject jsonObject = JSON.parseObject(s);
                        String skuId = jsonObject.getString("sku_id");
                        Long ts = jsonObject.getLong("ts");
                        if (skuId != null && ts != null) {
                            jsonObject.put("ts", ts * 1000);
                            jsonObject.put("shop_id",new Random().nextInt(10)+1);
                            collector.collect(jsonObject);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }


            // 验证JSON格式的工具方法
            private boolean isJSONValid(String json) {
                try {
                    JSON.parse(json);
                    return true;
                } catch (JSONException e) {
                    return false;
                }
            }
        });
//        etlStream.print("sssssssssssssssssssssssss");
        // 2.添加水位线
        SingleOutputStreamOperator<JSONObject> wmStream = etlStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getLong("ts");
            }
        }));
        // 3.分组
        KeyedStream<JSONObject, String> keyedStream = wmStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("id");
            }
        });

        keyedStream.print("hhhhhhhhhhhhhhhhhhhhhhh");
        // 4.防止下游重复计算
        SingleOutputStreamOperator<TradeSkuOrderBean> processStream = keyedStream.map(new MapFunction<JSONObject, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(JSONObject jsonObject) throws Exception {
                System.out.println("Processing data: " + jsonObject.toJSONString());

                try {
                    // 取出数据里面的度量值
                    BigDecimal splitTotalAmount = jsonObject.getBigDecimal("split_total_amount");
                    BigDecimal splitActivityAmount = jsonObject.getBigDecimal("split_activity_amount");
                    BigDecimal splitCouponAmount = jsonObject.getBigDecimal("split_coupon_amount");


                    // 检查order_price是否存在，如果不存在则设置默认值
                    BigDecimal orderPrice = jsonObject.getBigDecimal("order_price");
                    if (orderPrice == null) {
                        System.out.println("order_price is null, using zero as default");
                        orderPrice = BigDecimal.ZERO;
                    }

                    BigDecimal sku_num = jsonObject.getBigDecimal("sku_num");
                    if (sku_num == null) {
                        System.out.println("sku_num is null, using 1 as default");
                        sku_num = BigDecimal.ONE;
                    }

                    String shopId = jsonObject.getString("shop_id");
                    if (shopId == null) {
                        System.out.println("shop_id is null, using default value");
                        shopId = "0"; // 设置默认值
                    }

                    // 计算原始金额
                    BigDecimal originalAmount = orderPrice.multiply(sku_num);

                    // 构建输出对象 - 使用默认值而不是状态比较
                    String skuId = jsonObject.getString("sku_id");
                    String skuName = jsonObject.getString("sku_name");
                    String id = jsonObject.getString("id");
                    Long ts = jsonObject.getLong("ts");

                    // 直接使用当前值，不与之前的状态比较
                    TradeSkuOrderBean bean = TradeSkuOrderBean.builder()
                            .orderDetailId(id)
                            .skuName(skuName)
                            .ts(ts)
                            .skuId(skuId)
                            .originalAmount(originalAmount)
                            .activityReduceAmount(splitActivityAmount)
                            .couponReduceAmount(splitCouponAmount)
                            .orderAmount(splitTotalAmount)
                            .shopId(shopId)
                            .build();

                    System.out.println("Mapped to bean: " + bean);
                    return bean;
                } catch (Exception e) {
                    System.err.println("Error mapping element: " + jsonObject.toJSONString());
                    e.printStackTrace();
                    throw e; // 重新抛出异常，让Flink处理
                }
            }
        });

// 对processStream添加打印
//        processStream.print("Processed Bean: ");


        // 5.以sku_id分组
        KeyedStream<TradeSkuOrderBean, String> KeyedStream = processStream.keyBy(new KeySelector<TradeSkuOrderBean, String>() {
            @Override
            public String getKey(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                String skuId = tradeSkuOrderBean.getSkuId();
                if (skuId == null) {
                    System.err.println("skuId is null, using default key");
                    return "null_sku"; // 使用默认键而不是返回null
                }
                return skuId;
            }
        });
        keyedStream.print("Keyed Stream: ");
        // 6.聚合数据
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceStream = KeyedStream.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setCouponReduceAmount(value1.getCouponReduceAmount().add(value2.getCouponReduceAmount()));
                        value1.setActivityReduceAmount(value1.getActivityReduceAmount().add(value2.getActivityReduceAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                }, new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>.Context context, Iterable<TradeSkuOrderBean> iterable, Collector<TradeSkuOrderBean> collector) throws Exception {
                        // 为了取时间窗口
                        TimeWindow window = context.window();
                        long start = window.getStart();
                        long end = window.getEnd();
                        for (TradeSkuOrderBean tradeSkuOrderBean : iterable) {
                            tradeSkuOrderBean.setStt(DateFormatUtil.tsToDateTime(start));
                            tradeSkuOrderBean.setEdt(DateFormatUtil.tsToDateTime(end));
                            tradeSkuOrderBean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                            collector.collect(tradeSkuOrderBean);
                        }
                    }
                });
        ;
        reduceStream.print("Reduce Stream: ");
        // 7.关联维度
        SingleOutputStreamOperator<TradeSkuOrderBean> mapStream = reduceStream.map(new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
            Connection hbaseConnect;

            @Override
            public void open(Configuration parameters) throws Exception {
                hbaseConnect = HbaseUtil.getHbaseConnect();

            }

            @Override
            public void close() throws Exception {
                hbaseConnect.close();
            }

            @Override
            public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                // 先关联SKU
                String skuId = tradeSkuOrderBean.getSkuId();
                JSONObject dimSkuInfo = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId);
                String spuId = dimSkuInfo.getString("spu_id");
                String category3Id = dimSkuInfo.getString("category3_id");

                tradeSkuOrderBean.setSpuId(spuId);
                tradeSkuOrderBean.setCategory3Id(category3Id);
                tradeSkuOrderBean.setTrademarkId(dimSkuInfo.getString("tm_id"));

                String tmId = tradeSkuOrderBean.getTrademarkId();
                JSONObject dimBaseTrademark = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_trademark",tmId);
                tradeSkuOrderBean.setTrademarkName(dimBaseTrademark.getString("tm_name"));


                // 在关联spu
                JSONObject dimSpuInfo = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_spu_info", spuId);
                tradeSkuOrderBean.setSpuName(dimSpuInfo.getString("spu_name"));


                // 在关联C3
                JSONObject dimBaseCategory3 = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_category3", category3Id);
                tradeSkuOrderBean.setCategory3Name(dimBaseCategory3.getString("name"));
                String category2Id = dimBaseCategory3.getString("category2_id");
                tradeSkuOrderBean.setCategory2Id(dimBaseCategory3.getString("category2_id"));


                // 在关联C2
                JSONObject dimBaseCategory2 = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_category2", category2Id);
                String category1Id = dimBaseCategory2.getString("category1_id");
                tradeSkuOrderBean.setCategory2Name(dimBaseCategory2.getString("name"));
                tradeSkuOrderBean.setCategory1Id(dimBaseCategory2.getString("category1_id"));

                // 在关联C1
                JSONObject dimBaseCategory1 = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_category1", category1Id);
                tradeSkuOrderBean.setCategory1Name(dimBaseCategory1.getString("name"));

//                String shopId = tradeSkuOrderBean.getShopId();
//                tradeSkuOrderBean.setShopId(shopId);
//                String shopName = tradeSkuOrderBean.getShopName();
//                tradeSkuOrderBean.setShopName(shopName);

                return tradeSkuOrderBean;
            }
        });
        mapStream.print("Map Stream: ");

        // 8.写到Doris
        mapStream
                .map(new DorisMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_TRADE_SKU_ORDER_WINDOW));


    }
}
